#include <iostream>
#include <fstream>
#include <memory>
#include <thread>
#include <chrono>
#include <atomic>
#include <sstream>

#include "qmdb.monitor.pb.h"
#include <google/protobuf/io/zero_copy_stream.h>
#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
#include <google/protobuf/io/gzip_stream.h>
#include <WinSock2.h>

#ifdef _DEBUG_
#pragma comment(lib, "libprotobufd.lib")
#pragma comment(lib, "libprotocd.lib")
#pragma comment(lib, "zlibd.lib")
#else
#pragma comment(lib, "libprotobuf.lib")
#pragma comment(lib, "libprotoc.lib")
#pragma comment(lib, "zlib.lib")
#endif //_DEBUG_

#pragma comment(lib, "Ws2_32.lib")

using namespace std;
using namespace google::protobuf;
using namespace google::protobuf::io;

void decodelargermsg(char* recvBuff, int& totalCount, int& msgtype, int& bodyLen, int headLen, int i, FILE *pfile, int& recvMsg)
{
    while (totalCount >= (int)bodyLen + headLen)
    {
        if (msgtype != -1)
        {
            google::protobuf::io::ArrayInputStream arrayInputStream(recvBuff + headLen, bodyLen);
            qmdb::monitor msg1;
            bool isSuccess = false;
            if (i % 2 == 1)
            {
                google::protobuf::io::GzipInputStream gzip_in(&arrayInputStream);
                isSuccess = msg1.ParseFromZeroCopyStream(&gzip_in);
            }
            else
            {
                isSuccess = msg1.ParseFromZeroCopyStream(&arrayInputStream);
            }
            recvMsg++;
            if (recvMsg == 1)
            {
                cout << "msg1.ByteSize:" << msg1.ByteSize() << endl;
            }
            if (!isSuccess)
            {
                cerr << "parse failed." << recvMsg << endl;
                cerr << "totalCount:" << totalCount << " bodyLen:" << bodyLen << endl;
                cerr << "body:" << endl;
                for (int i = headLen; i < headLen + headLen; i++)
                {
                    cout << std::hex << (unsigned int)(recvBuff[i]);
                }
                cout << std::dec << endl;
                exit(-1);
            }


            fwrite(msg1.msg().c_str(), msg1.msg().length(), 1, pfile);
            if (recvMsg % 1000 == 0)
            {
                fflush(pfile);
            }
        }

        if (totalCount > (int)bodyLen + headLen)
        {
            std::copy(recvBuff + bodyLen + headLen, recvBuff + totalCount, recvBuff);
        }

        totalCount -= ((int)bodyLen + headLen);
        bodyLen = 0;
        if (totalCount >= headLen)
        {
            int netdata = 0;
            memcpy(&netdata, recvBuff, sizeof(netdata));
            msgtype = ntohl(netdata);
            memcpy(&netdata, recvBuff + 4, sizeof(netdata));
            bodyLen = ntohl(netdata);
        }
        else
        {
            break;
        }
    }
}

void decodesimplemsg(char* recvBuff, int& totalCount, int& msgtype, int &bodyLen, int headLen, int i, FILE *pfile, int& recvMsg)
{
    while (totalCount >= (int)bodyLen + headLen)
    {
        if (msgtype != -1)
        {
            google::protobuf::io::ArrayInputStream arrayInputStream(recvBuff + headLen, bodyLen);
            qmdb::simplemsg msg1;
            bool isSuccess = false;
            if (i % 2 == 1)
            {
                google::protobuf::io::GzipInputStream gzip_in(&arrayInputStream);
                isSuccess = msg1.ParseFromZeroCopyStream(&gzip_in);
            }
            else
            {
                isSuccess = msg1.ParseFromZeroCopyStream(&arrayInputStream);
            }
            recvMsg++;
            if (recvMsg == 1)
            {
                cout << "msg1.ByteSize:" << msg1.ByteSize() << endl;
            }
            if (!isSuccess)
            {
                cerr << "parse failed." << recvMsg << endl;
                cerr << "totalCount:" << totalCount << " bodyLen:" << bodyLen << endl;
                cerr << "body:" << endl;
                for (int i = headLen; i < headLen + headLen; i++)
                {
                    cout << std::hex << (unsigned int)(recvBuff[i]);
                }
                cout << std::dec << endl;
                exit(-1);
            }


            fwrite(msg1.content().c_str(), msg1.content().length(), 1, pfile);
            if (recvMsg % 1000 == 0)
            {
                fflush(pfile);
            }
        }

        if (totalCount > (int)bodyLen + headLen)
        {
            std::copy(recvBuff + bodyLen + headLen, recvBuff + totalCount, recvBuff);
        }

        totalCount -= ((int)bodyLen + headLen);
        bodyLen = 0;
        if (totalCount >= headLen)
        {
            int netdata = 0;
            memcpy(&netdata, recvBuff, sizeof(netdata));
            msgtype = ntohl(netdata);
            memcpy(&netdata, recvBuff + 4, sizeof(netdata));
            bodyLen = ntohl(netdata);
        }
        else
        {
            break;
        }
    }
}

void decodequerynextmsg(char* recvBuff, int& totalCount, int& msgtype, int &bodyLen, int headLen, int i, FILE *pfile, int& recvMsg)
{
    while (totalCount >= (int)bodyLen + headLen)
    {
        if (msgtype != -1)
        {
            google::protobuf::io::ArrayInputStream arrayInputStream(recvBuff + headLen, bodyLen);
            qmdb::querynextmsg msg1;
            bool isSuccess = false;
            if (i % 2 == 1)
            {
                google::protobuf::io::GzipInputStream gzip_in(&arrayInputStream);
                isSuccess = msg1.ParseFromZeroCopyStream(&gzip_in);
            }
            else
            {
                isSuccess = msg1.ParseFromZeroCopyStream(&arrayInputStream);
            }
            recvMsg++;
            if (recvMsg == 1)
            {
                cout << "msg1.ByteSize:" << msg1.ByteSize() << endl;
            }
            if (!isSuccess)
            {
                cerr << "parse failed." << recvMsg << endl;
                cerr << "totalCount:" << totalCount << " bodyLen:" << bodyLen << endl;
                cerr << "body:" << endl;
                for (int i = headLen; i < headLen + headLen; i++)
                {
                    cout << std::hex << (unsigned int)(recvBuff[i]);
                }
                cout << std::dec << endl;
                exit(-1);
            }

            int recordCount = msg1.records_size();
            for (int index = 0; index < recordCount; index++)
            {
                const qmdb::recordmsg& record = msg1.records(index);
                int colCount = record.columns_size();
                for (int colIndex = 0; colIndex < colCount; colIndex++)
                {
                    if (record.columns(colIndex).name() == "comment" && record.columns(colIndex).type() == 1)
                    {
                        fwrite(record.columns(colIndex).strvalue().c_str(), record.columns(colIndex).strvalue().length(), 1, pfile);
                    }
                }
            }
            
            if (recvMsg % 1000 == 0)
            {
                fflush(pfile);
            }
        }

        if (totalCount > (int)bodyLen + headLen)
        {
            std::copy(recvBuff + bodyLen + headLen, recvBuff + totalCount, recvBuff);
        }

        totalCount -= ((int)bodyLen + headLen);
        bodyLen = 0;
        if (totalCount >= headLen)
        {
            int netdata = 0;
            memcpy(&netdata, recvBuff, sizeof(netdata));
            msgtype = ntohl(netdata);
            memcpy(&netdata, recvBuff + 4, sizeof(netdata));
            bodyLen = ntohl(netdata);
        }
        else
        {
            break;
        }
    }
}

struct simplestruct
{
    int cmdcode = 1;
    int sessionid = 1000;
    int sequence = -1;
    string content;
};

void decodesimplestruct(char* recvBuff, int& totalCount, int& msgtype, int& bodyLen, int headLen, int i, FILE *pfile, int& recvMsg)
{
    while (totalCount >= (int)bodyLen + headLen)
    {
        if (msgtype != -1)
        {
            simplestruct msg1;
            recvMsg++;

            int netdata = 0;
            memcpy(&netdata, recvBuff + headLen, sizeof(netdata));
            msg1.cmdcode = ntohl(netdata);
            memcpy(&netdata, recvBuff + headLen +4, sizeof(netdata));
            msg1.sequence = ntohl(netdata);
            memcpy(&netdata, recvBuff + headLen + 8, sizeof(netdata));
            msg1.sessionid = ntohl(netdata);
            memcpy(&netdata, recvBuff + headLen + 12, sizeof(netdata));
            int strLen = ntohl(netdata);
            msg1.content = string(recvBuff + headLen + 16, strLen);

            if (recvMsg == 1)
            {
                cout << "msg1.ByteSize:" << 16 + msg1.content.length() << endl;
            }

            fwrite(msg1.content.c_str(), msg1.content.length(), 1, pfile);
            if (recvMsg % 1000 == 0)
            {
                fflush(pfile);
            }
        }

        if (totalCount > (int)bodyLen + headLen)
        {
            std::copy(recvBuff + bodyLen + headLen, recvBuff + totalCount, recvBuff);
        }

        totalCount -= ((int)bodyLen + headLen);
        bodyLen = 0;
        if (totalCount >= headLen)
        {
            int netdata = 0;
            memcpy(&netdata, recvBuff, sizeof(netdata));
            msgtype = ntohl(netdata);
            memcpy(&netdata, recvBuff + 4, sizeof(netdata));
            bodyLen = ntohl(netdata);
        }
        else
        {
            break;
        }
    }
}


struct columninfo
{
    int type;
    string name;
    int intvalue;
    int64 longvalue;
    string strvalue;
};

struct recordinfo
{
    vector<columninfo> columns;
};

struct querynextmsg
{
    vector<recordinfo> records;
};

void decodequerynextstruct(char* recvBuff, int& totalCount, int& msgtype, int& bodyLen, int headLen, int i, FILE *pfile, int& recvMsg)
{
    while (totalCount >= (int)bodyLen + headLen)
    {
        if (msgtype != -1)
        {
            querynextmsg msg1;
            recvMsg++;

            int netdata = 0;
            int readPos = headLen;
            memcpy(&netdata, recvBuff + readPos, sizeof(netdata));
            readPos += sizeof(netdata);

            int recordCount = ntohl(netdata);
            for (int index = 0; index < recordCount; index++)
            {
                memcpy(&netdata, recvBuff + readPos, sizeof(netdata));
                readPos += sizeof(netdata);
                int colCount = ntohl(netdata);
                recordinfo record;
                for (int colIndex = 0; colIndex < colCount; colIndex++)
                {
                    columninfo colInfo;
                    memcpy(&netdata, recvBuff + readPos, sizeof(netdata));
                    readPos += sizeof(netdata);
                    int nameLen = ntohl(netdata);
                    colInfo.name = string(recvBuff + readPos, nameLen);
                    readPos += nameLen;

                    memcpy(&netdata, recvBuff + readPos, sizeof(netdata));
                    readPos += sizeof(netdata);
                    colInfo.type = ntohl(netdata);

                    switch (colInfo.type)
                    {
                    case 0://int
                        memcpy(&netdata, recvBuff + readPos, sizeof(netdata));
                        readPos += sizeof(netdata);
                        colInfo.intvalue = ntohl(netdata);
                        break;
                    case 1://string
                    {
                        memcpy(&netdata, recvBuff + readPos, sizeof(netdata));
                        readPos += sizeof(netdata);
                        int vallen = ntohl(netdata);
                        colInfo.strvalue = string(recvBuff + readPos, vallen);
                        readPos += vallen;
                    }
                        break;
                    default:
                        break;
                    }
                    record.columns.push_back(std::move(colInfo));
                }
                msg1.records.push_back(std::move(record));
            }

            if (readPos -8 != bodyLen)
            {
                cerr << "readPos-8=" << readPos - 8 << " bodyLen=" << bodyLen << endl;
                exit(-1);
            }

            if (recvMsg == 1)
            {
                cout << "msg1.ByteSize:" << readPos - 8 << endl;
            }

            for (auto& item:msg1.records)
            {
                for (auto& colItem: item.columns)
                {
                    if (colItem.name == "comment" && colItem.type == 1)
                    {
                        fwrite(colItem.strvalue.c_str(), colItem.strvalue.length(), 1, pfile);
                    }
                }
            }
            
            if (recvMsg % 1000 == 0)
            {
                fflush(pfile);
            }
        }

        if (totalCount > (int)bodyLen + headLen)
        {
            std::copy(recvBuff + bodyLen + headLen, recvBuff + totalCount, recvBuff);
        }

        totalCount -= ((int)bodyLen + headLen);
        bodyLen = 0;
        if (totalCount >= headLen)
        {
            int netdata = 0;
            memcpy(&netdata, recvBuff, sizeof(netdata));
            msgtype = ntohl(netdata);
            memcpy(&netdata, recvBuff + 4, sizeof(netdata));
            bodyLen = ntohl(netdata);
        }
        else
        {
            break;
        }
    }
}

int main(int argc, char* argv[])
{
    WSADATA wsaData;
    int iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
    if (iResult != NO_ERROR) {
        cerr << "WSAStartup function failed with error: " << iResult << endl;
        return 1;
    }

    SOCKET ConnectSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (INVALID_SOCKET == ConnectSocket)
    {
        cerr << "invalid socket." << __LINE__ << endl;
        return -1;
    }

    sockaddr_in clientService;
    clientService.sin_family = AF_INET;
    clientService.sin_addr.s_addr = inet_addr("10.45.17.152");
    clientService.sin_port = htons(9000);
    iResult = connect(ConnectSocket, (SOCKADDR *)& clientService, sizeof(clientService));
    if (iResult == SOCKET_ERROR) 
    {
        cerr << "connect function failed with error: " << errno << endl;
        iResult = closesocket(ConnectSocket);
        if (iResult == SOCKET_ERROR)
            cerr << "closesocket function failed with error: " << errno << endl;
        WSACleanup();
        return 1;
    }

    vector<char> recvBuff(1024*64, 0);
    unique_ptr<char[]> gzipbuff(new char[1024*32]);
    memset(gzipbuff.get(), 0, 1024);
    int readCount = 0;
    int totalCount = 0;
    int bodyLen = 0;
    int msgtype = 0;
    const int headLen = 8;

    for (char i = 0; i < 10; i += 1)
    {
        int recvMsg = 0;
        bool toSwitch = false;
        int totalRecv = 0;
        msgtype = 0;
        stringstream formatstream;
        formatstream << "D:/content" << (int)i << ".txt";
        cout << "writetofile:" << formatstream.str() << endl;
        unique_ptr<FILE, decltype(fclose)*> contentfile(fopen(formatstream.str().c_str(), "wb"), fclose);
        if (nullptr == contentfile.get())
        {
            cerr << "open file content.txt faied." << endl;
            return -1;
        }

        int sendbytes = send(ConnectSocket, &i, 1, 0);
        if (-1 == sendbytes)
        {
            cerr << "send failed errno:" << errno << " errmsg:" << strerror(errno) << endl;
            return -1;
        }
        auto startpoint = chrono::steady_clock::now();
        while ((msgtype != -1)
            && (readCount = ::recv(ConnectSocket, (char*)(&*recvBuff.begin()) + totalCount, recvBuff.size() - 1 - totalCount, 0)) > 0)
        {
            totalCount += readCount;
            totalRecv += readCount;
            if (totalCount < headLen)
            {
                continue;
            }

            if (bodyLen == 0 && totalCount >= headLen)
            {
                int netdata = 0;
                memcpy(&netdata, (char*)(&*recvBuff.begin()), sizeof(netdata));
                msgtype = ntohl(netdata);
                memcpy(&netdata, (char*)(&*recvBuff.begin()) + 4, sizeof(netdata));
                bodyLen = ntohl(netdata);
            }

            if (recvBuff.size() < (int)bodyLen + headLen)
            {
                recvBuff.resize((int)bodyLen + headLen);
                cout << "resize" << endl;
            }

            switch (msgtype)
            {
            case 0:
            case 1:
                decodelargermsg(&*recvBuff.begin(), totalCount, msgtype, bodyLen, headLen, i, contentfile.get(), recvMsg);
                break;
            case 2:
            case 3:
                decodesimplemsg(&*recvBuff.begin(), totalCount, msgtype, bodyLen, headLen, i, contentfile.get(), recvMsg);
                break;
            case 4:
            case 5:
                decodesimplestruct(&*recvBuff.begin(), totalCount, msgtype, bodyLen, headLen, i, contentfile.get(), recvMsg);
                break;
            case 6:
            case 7:
                decodequerynextmsg(&*recvBuff.begin(), totalCount, msgtype, bodyLen, headLen, i, contentfile.get(), recvMsg);
                break;
            case 8:
            case 9:
                decodequerynextstruct(&*recvBuff.begin(), totalCount, msgtype, bodyLen, headLen, i, contentfile.get(), recvMsg);
                break;
            default:
                break;
            }
        }
        auto endpoint = chrono::steady_clock::now();
        cout << (int)i << "-->recvbytes:" << totalRecv << " recvmsgs:" << recvMsg <<" usedtime(milliseconds):" << chrono::duration_cast<chrono::milliseconds>(endpoint - startpoint).count() << endl;
    }
    cout << "exit ..." << endl;
    WSACleanup();

    return 0;
}